//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// nested_loop_join_executor.cpp
//
// Identification: src/execution/nested_loop_join_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "execution/executors/nested_loop_join_executor.h"
#include <cstddef>
#include <utility>
#include <vector>
#include "binder/table_ref/bound_join_ref.h"
#include "common/exception.h"
#include "common/logger.h"
#include "common/macros.h"
#include "common/rid.h"
#include "fmt/format.h"
#include "storage/table/tuple.h"
#include "type/value.h"
#include "type/value_factory.h"

namespace bustub {

NestedLoopJoinExecutor::NestedLoopJoinExecutor(ExecutorContext *exec_ctx, const NestedLoopJoinPlanNode *plan,
                                               std::unique_ptr<AbstractExecutor> &&left_executor,
                                               std::unique_ptr<AbstractExecutor> &&right_executor)
    : AbstractExecutor(exec_ctx),
      plan_(plan),
      left_executor_(std::move(left_executor)),
      right_executor_(std::move(right_executor)),
      left_schema_(&left_executor_->GetOutputSchema()),
      right_schema_(&right_executor_->GetOutputSchema()) {
  if (plan->GetJoinType() != JoinType::LEFT && plan->GetJoinType() != JoinType::INNER) {
    // Note for 2023 Spring: You ONLY need to implement left join and inner join.
    throw bustub::NotImplementedException(fmt::format("join type {} not supported", plan->GetJoinType()));
  }
  // if left join ,  handle NULL situations
  matched_ = (plan->join_type_ == JoinType::INNER);
}

void NestedLoopJoinExecutor::Init() {
  left_executor_->Init();
  right_executor_->Init();
  matched_ = (plan_->join_type_ == JoinType::INNER);
  first_ = true;
  values_.clear();
}

auto NestedLoopJoinExecutor::Next(Tuple *tuple, RID *rid) -> bool {
  if (first_) {
    values_.reserve(left_schema_->GetColumnCount() + right_schema_->GetColumnCount());
    if (!left_executor_->Next(&left_tuple_, &dummy_rid_)) {
      return false;
    }
    first_ = false;
    // copy left tuple into
    for (size_t i = 0; i < left_schema_->GetColumnCount(); ++i) {
      values_.push_back(left_tuple_.GetValue(left_schema_, i));
    }
  }
  Tuple right_tuple{};
  while (true) {
    if (!right_executor_->Next(&right_tuple, &dummy_rid_)) {
      if (!matched_) {
        // padding with NULL
        for (size_t i = 0; i < right_schema_->GetColumnCount(); ++i) {
          values_.push_back(ValueFactory::GetNullValueByType(right_schema_->GetColumn(i).GetType()));
        }
        BUSTUB_ASSERT(values_.size() == GetOutputSchema().GetColumnCount(), "wrong size");
        // emit Tuple
        *tuple = {values_, &GetOutputSchema()};
        matched_ = true;
        return true;
      }
      if (!left_executor_->Next(&left_tuple_, &dummy_rid_)) {
        return false;
      }
      // copy left tuple into
      values_.clear();
      for (size_t i = 0; i < left_schema_->GetColumnCount(); ++i) {
        values_.push_back(left_tuple_.GetValue(left_schema_, i));
      }
      right_executor_->Init();
      if (plan_->join_type_ == JoinType::LEFT) {
        matched_ = false;
      }
    } else {
      auto res = plan_->Predicate()->EvaluateJoin(&left_tuple_, plan_->GetLeftPlan()->OutputSchema(), &right_tuple,
                                                  plan_->GetRightPlan()->OutputSchema());
      if (res.GetAs<bool>()) {
        // copy right tuple into values
        for (size_t i = 0; i < right_schema_->GetColumnCount(); ++i) {
          values_.push_back(right_tuple.GetValue(right_schema_, i));
        }
        BUSTUB_ASSERT(values_.size() == GetOutputSchema().GetColumnCount(), "1");
        *tuple = {values_, &GetOutputSchema()};
        values_.resize(left_schema_->GetColumnCount());
        values_.reserve(GetOutputSchema().GetColumnCount());
        matched_ = true;
        return true;
      }
    }
  }
}

}  // namespace bustub
